﻿using System;
using System.Collections.Generic;
using System.Linq;
using System.Text;
using System.Threading.Tasks;
using System.Net;
using System.Threading;

namespace QuickStart
{
    using ENode.Configurations;
    using ENode.Infrastructure;
    using ENode.Infrastructure.Impl;
    using EQueue.Configurations;
    using ENode.EQueue;
    using ENode.Eventing;
    using EQueue.Clients.Producers;
    using EQueue.Clients.Consumers;
    using ECommon.Utilities;
    using ECommon.Scheduling;
    using ECommon.Log4Net;
    using ECommon.Components;
    using ENode.Commanding;
    using ECommon.Logging;

    using Domain.Accounts;
    using Commands.Accounts;
    using Infrastructure;
    using Messages;
    using ProcessManagers;
    using Denormalizers.Ef;
    using ECommon.Socketing;

    public static class ENodeExtensions
    {
        public static ENodeConfiguration RegisterAllTypeCodes(this ENodeConfiguration enodeConfiguration)
        {
            var provider = ObjectContainer.Resolve<ITypeCodeProvider>() as DefaultTypeCodeProvider;


            //aggregates
            provider.RegisterType<Account>(AllTypeCodes.Account);
            provider.RegisterType<Domain.Posts.Post>(AllTypeCodes.Post);
            provider.RegisterType<Domain.Replies.Reply>(AllTypeCodes.Reply);
            provider.RegisterType<Domain.Sections.Section>(AllTypeCodes.Section);




            //commands
            provider.RegisterType<RegisterNewAccountCommand>(AllTypeCodes.RegisterNewAccountCommand);
            provider.RegisterType<SignInAfterCommand>(AllTypeCodes.SignInAfterCommand);
            provider.RegisterType<SignOutAfterCommand>(AllTypeCodes.SignOutAfterCommand);


            provider.RegisterType<Commands.Sections.CreateSectionCommand>(AllTypeCodes.CreateSectionCommand);
            provider.RegisterType<Commands.Sections.ChangeSectionNameCommand>(AllTypeCodes.ChangeSectionNameCommand);

            provider.RegisterType<Commands.Posts.CreatePostCommand>(AllTypeCodes.CreatePostCommand);
            provider.RegisterType<Commands.Posts.UpdatePostCommand>(AllTypeCodes.UpdatePostCommand);
            provider.RegisterType<Commands.Posts.AcceptNewReplyCommand>(AllTypeCodes.AcceptNewReplyCommand);

            provider.RegisterType<Commands.Replies.CreateReplyCommand>(AllTypeCodes.CreateReplyCommand);
            provider.RegisterType<Commands.Replies.ChangeReplyBodyCommand>(AllTypeCodes.ChangeReplyBodyCommand);

            //events
            provider.RegisterType<NewAccountRegisteredEvent>(AllTypeCodes.NewAccountRegisteredEvent);
            provider.RegisterType<SignInAfterEvent>(AllTypeCodes.SignInAfterEvent);
            provider.RegisterType<SignOutAfterEvent>(AllTypeCodes.SignOutAfterEvent);



            provider.RegisterType<Domain.Sections.SectionCreatedEvent>(AllTypeCodes.SectionCreatedEvent);
            provider.RegisterType<Domain.Sections.SectionNameChangedEvent>(AllTypeCodes.SectionNameChangedEvent);

            provider.RegisterType<Domain.Posts.PostCreatedEvent>(AllTypeCodes.PostCreatedEvent);
            provider.RegisterType<Domain.Posts.PostUpdatedEvent>(AllTypeCodes.PostUpdatedEvent);
            provider.RegisterType<Domain.Posts.PostReplyStatisticInfoChangedEvent>(AllTypeCodes.PostReplyStatisticInfoChangedEvent);

            provider.RegisterType<Domain.Replies.ReplyCreatedEvent>(AllTypeCodes.ReplyCreatedEvent);
            provider.RegisterType<Domain.Replies.ReplyBodyChangedEvent>(AllTypeCodes.ReplyBodyChangedEvent);

            //application message
            //provider.RegisterType<SignInRejectedMessage>(300);

            //publishable exceptions
            //provider.RegisterType<DisabledException>(500);

            //application message, domain event, or exception handlers
            //provider.RegisterType<OAuthProcessManager>(620);
            provider.RegisterType<AccountDenormalizer>(AllTypeCodes.AccountDenormalizer);
            provider.RegisterType<PostDenormalizer>(AllTypeCodes.PostDenormalizer);
            provider.RegisterType<ReplyDenormalizer>(AllTypeCodes.ReplyDenormalizer);
            provider.RegisterType<SectionDenormalizer>(AllTypeCodes.SectionDenormalizer);


            provider.RegisterType<CreateReplyProcessManager>(AllTypeCodes.CreateReplyProcessManager);

            return enodeConfiguration;
        }


        public static ENodeConfiguration UseEQueue(this ENodeConfiguration enodeConfiguration)
        {
            var configuration = enodeConfiguration.GetCommonConfiguration();
            configuration.RegisterEQueueComponents();

            BindQueue(configuration);
            return enodeConfiguration;
        }
         


        public static ENodeConfiguration StartEQueue(this ENodeConfiguration enodeConfiguration)
        {
            InitializeCommandService();

            _eventConsumer.Start();
            _commandConsumer.Start();
            _messageConsumer.Start();
            _exceptionConsumer.Start();

            _exceptionPublisher.Start();
            _eventPublisher.Start();
            _commandService.Start();

            WaitAllConsumerLoadBalanceComplete();
            return enodeConfiguration;
        }


        private static void InitializeCommandService()
        {
            ObjectContainer.Resolve<ILockService>().AddLockKey(typeof(Domain.Accounts.AccountIndex).Name);

            var _logger = ObjectContainer.Resolve<ILoggerFactory>().Create(typeof(ENodeExtensions).Name);
            _logger.Info("Command service initialized.");
        }



        private static CommandService _commandService;// 或当前进程内部不需要发送cmd，就不需要这个了。
        private static CommandConsumer _commandConsumer;

        private static DomainEventPublisher _eventPublisher;
        private static DomainEventConsumer _eventConsumer;

        private static ApplicationMessageConsumer _messageConsumer;

        private static PublishableExceptionPublisher _exceptionPublisher;
        private static PublishableExceptionConsumer _exceptionConsumer;

        private static void BindQueue(ECommon.Configurations.Configuration configuration)
        {
            // 生产者
            var producerSetting = new ProducerSetting
            {
                BrokerAddress =
                    new IPEndPoint(SocketUtils.GetLocalIPV4(), ConfigSettings.BrokerProducerPort)
            };
            _commandService = new CommandService(
                null, "CommandService", producerSetting);
            configuration.SetDefault<ICommandService,
                CommandService>(_commandService);

            _eventPublisher = new DomainEventPublisher(
                "DomainEventPublisher", producerSetting);
            configuration.SetDefault<IMessagePublisher<DomainEventStreamMessage>,
                DomainEventPublisher>(_eventPublisher);

            _exceptionPublisher = new PublishableExceptionPublisher(
                "PublishableExceptionPublisher", producerSetting);
            configuration.SetDefault<IMessagePublisher<IPublishableException>,
                PublishableExceptionPublisher>(_exceptionPublisher);

            //  消费者
            var consumerSetting = new ConsumerSetting
            {
                BrokerAddress =
                    new IPEndPoint(SocketUtils.GetLocalIPV4(), ConfigSettings.BrokerConsumerPort)
            };
            _commandConsumer = new CommandConsumer(
                "CommandConsumer",
                "CommandConsumerGroup",
                consumerSetting)
            .Subscribe(Topics.DefaultCommand);

            _eventConsumer = new DomainEventConsumer(
                "EventConsumer",
                "EventConsumerGroup",
                consumerSetting)
            .Subscribe(Topics.DetaultEvent);

            _messageConsumer = new ApplicationMessageConsumer(
                "MessageConsumer",
                "MessageConsumerGroup",
                consumerSetting)
            .Subscribe(Topics.DetaultApplicationMessage);

            _exceptionConsumer = new PublishableExceptionConsumer(
                "ExceptionConsumer",
                "ExceptionConsumerGroup",
                consumerSetting)
            .Subscribe(Topics.DetaultPublishableException);

        }


        /// <summary>
        /// 等待各消费者做好负载均衡
        /// </summary>
        private static void WaitAllConsumerLoadBalanceComplete()
        {
            var logger = ObjectContainer.Resolve<ILoggerFactory>().Create(typeof(ENodeExtensions).Name);
            var scheduleService = ObjectContainer.Resolve<IScheduleService>();
            var waitHandle = new ManualResetEvent(false);
            logger.Info("Waiting for all consumer load balance complete, please wait for a moment...");
              scheduleService.StartTask("WaitAllConsumerLoadBalanceComplete", () =>
            {
                var eventConsumerAllocatedQueues = _eventConsumer.Consumer.GetCurrentQueues();
                var commandConsumerAllocatedQueues = _commandConsumer.Consumer.GetCurrentQueues();
                if (eventConsumerAllocatedQueues.Count() == 4 &&
                    commandConsumerAllocatedQueues.Count() == 4)
                {
                    waitHandle.Set();
                }
            }, 1000, 1000);

            waitHandle.WaitOne();
            scheduleService.StopTask("WaitAllConsumerLoadBalanceComplete");
            logger.Info("All consumer load balance completed.");
        }

    }
}
